package top.yaofengqiao.springcloudsimple.payment.biz.trigger.listener;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.messaging.Message;
import top.yaofengqiao.springcloudsimple.payment.biz.domain.pay.model.entity.PaymentOrderEntity;
import top.yaofengqiao.springcloudsimple.payment.biz.domain.pay.model.req.PaymentSuccessRequest;
import top.yaofengqiao.springcloudsimple.payment.biz.domain.pay.model.valobj.PaymentOrderStatus;
import top.yaofengqiao.springcloudsimple.payment.biz.domain.pay.service.IPaymentService;

import javax.annotation.Resource;
import java.time.LocalDateTime;

/**
 * @author yfq
 * @date 2024/6/24 10:17
 * @description 订单支付成功事务消息本地事务执行器
 */
@Slf4j
@RocketMQTransactionListener(rocketMQTemplateBeanName = "extRocketMQTemplate")
public class TransactionMessageListener implements RocketMQLocalTransactionListener {
    @Resource
    private IPaymentService paymentService;

    /**
     * executeLocalTransaction是半事务消息发送之后执行本地事务方法
     * 具体执行完本地事务后，可以在该方法中返回以下三种状态：
     * <p>
     * 1. LocalTransactionState.COMMIT_MESSAGE：提交事务，允许消费者消费该消息
     * <p>
     * 2. LocalTransactionState.ROLLBACK_MESSAGE：回滚事务，消息将被丢弃不允许消费。
     * <p>
     * 3. LocalTransactionState.UNKNOWN：暂时无法判断状态，等待固定时间以后Broker端根据回查规则向生产者进行消息回查。
     */
    @Override
    public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
        log.info("半事务消息发送成功，本地事务开始执行：{}", JSON.toJSONString(msg));
        String json = new String((byte[]) msg.getPayload());
        PaymentSuccessRequest request = JSON.parseObject(json, PaymentSuccessRequest.class);

        // 修改支付订单状态
        PaymentOrderEntity updateEntity = PaymentOrderEntity.builder()
                .payOrderNo(request.getOrderNo())
                .payAmount(request.getPayAmount())
                .payTime(request.getPayTime())
                .status(PaymentOrderStatus.SUCCESS.getStatus())
                .updateTime(LocalDateTime.now())
                .build();
        boolean result = paymentService.updatePaymentOrderByPayOrderNo(updateEntity);

        return result ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
    }

    /**
     * Broker向生产者检查本地事务的状态，如果本地事务为rollback则不会投递消息，只有事务为 commit时才会向下游投递消息
     */
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
        String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
        log.info("Broker检查事务消息状态：{}", transId);

        PaymentOrderStatus paymentOrderStatus = paymentService.queryPaymentOrderStatus(transId);

        if (paymentOrderStatus == PaymentOrderStatus.SUCCESS) return RocketMQLocalTransactionState.COMMIT;

        if (PaymentOrderStatus.TRADING == paymentOrderStatus) return RocketMQLocalTransactionState.UNKNOWN;

        return RocketMQLocalTransactionState.ROLLBACK;
    }
}
